package com.heima.recommend.kafka.listener;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.ArticleConstants;
import com.heima.common.constants.KafkaMessageConstants;
import com.heima.model.article.entity.ApArticle;
import com.heima.model.message.ArticleVisitStreamMess;
import com.heima.model.message.UpdateArticleMess;
import com.heima.recommend.dao.ApArticleDao;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;

/**
 * kafka监听器
 */
@Component
@Slf4j
public class HotArticleStreamListener {

    @Resource
    private ApArticleDao apArticleDao;

    @Resource
    private StringRedisTemplate stringRedisTemplate;


    @KafkaListener(topics = KafkaMessageConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC,groupId = "hot_article")
    public void wmNewsAutoScan(ConsumerRecord<String, String> record) {
        String articleId = record.key();
        String value = record.value();      //COLLECTION:2,LIKES:1,VIEWS:3,COMMENT:0
        log.info("接收到行为数据聚合结果, key:{},value:{}", articleId, value);
        //完成业务处理  更新数据库中的行为数量, 更新缓存分值
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        //获取其他属性封装到mess对象中
        String[] split = value.split(",");
        HashMap<String, Integer> map = new HashMap<>();
        for (String s : split) {
            String[] values = s.split(":");
            map.put(values[0], Integer.parseInt(values[1]));
        }
        mess.setCollect(map.get(UpdateArticleMess.UpdateArticleType.COLLECTION.toString()));
        mess.setComment(map.get(UpdateArticleMess.UpdateArticleType.COMMENT.toString()));
        mess.setLike(map.get(UpdateArticleMess.UpdateArticleType.LIKES.toString()));
        mess.setView(map.get(UpdateArticleMess.UpdateArticleType.VIEWS.toString()));
        log.info("接收到行为数据聚合,完成数据封装:{}", mess);

        //更新数据库的缓存数量，更新缓存分值
        ApArticle apArticle = apArticleDao.queryById(mess.getArticleId());
        if (apArticle == null) {
            return;
        }
        //更新行为数量
        updateArticle(mess, apArticle);
        //更新缓存分值
        Integer score = computeScore(mess);
        score = score * 3;
        //为每个频道缓存推荐文章分值数据
        stringRedisTemplate.opsForZSet().incrementScore(ArticleConstants.HOT_ARTICLE_PREFIX + apArticle.getChannelId(), apArticle.getId().toString(), score);
        //为推荐频道缓存推荐文章分值数据
        stringRedisTemplate.opsForZSet().incrementScore(ArticleConstants.HOT_ARTICLE_PREFIX + ArticleConstants.DEFAULT_TAG, apArticle.getId().toString(), score);
        //保存文章数据到Hash集合
        stringRedisTemplate.opsForHash().put(ArticleConstants.HOT_ARTICLE_HASH,apArticle.getId().toString(), JSON.toJSONString(apArticle));
    }

    /**
     * 分值计算
     * @param mess
     * @return
     */
    private Integer computeScore(ArticleVisitStreamMess mess) {
        int score = 0;
        if (mess.getLike() != 0) {
            score += mess.getLike() * ArticleConstants.HOT_ARTICLE_LIKE_WEIGHT;
        }

        if (mess.getView() != 0) {
            score += mess.getView() * ArticleConstants.HOT_ARTICLE_VIEW_WEIGHT;
        }

        if (mess.getComment() != 0) {
            score += mess.getComment() * ArticleConstants.HOT_ARTICLE_COMMENT_WEIGHT;
        }

        if (mess.getCollect() != 0) {
            score += mess.getCollect() * ArticleConstants.HOT_ARTICLE_COLLECT_WEIGHT;
        }
        return score;
    }

    /**
     * 更新文章行为数量
     *
     * @param mess
     * @param apArticle
     */
    private void updateArticle(ArticleVisitStreamMess mess, ApArticle apArticle) {
        apArticle.setCollection(apArticle.getCollection() + mess.getCollect());
        apArticle.setLikes(apArticle.getLikes() + mess.getLike());
        apArticle.setComment(apArticle.getComment() + mess.getComment());
        apArticle.setViews(apArticle.getViews() + mess.getView());
        apArticleDao.updateById(apArticle);
    }
}

